package kafka;

import com.mongodb.client.MongoCollection;
import mongo.MongoTest;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;
import org.bson.Document;
import util.DateUtil;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * kafka 线程类
 * USER: lintc 【lintiancong@zhuojianchina.com】
 * DATE: 2017-03-02 09:52
 */
public class KafkaRunnable implements Runnable {

    private static Logger LOG = Logger.getLogger(KafkaRunnable.class);

    private boolean isRun = true;

    private String topic;

    private String time = DateUtil.dateToString(new Date(), "yyyy-MM-dd");

    private LogMap logMap = new LogMap();

    // 处理列表
    private List<KafkaHandle> handles;

    /**
     * 当前日志抓取量
     */
    public Integer logNum = 0;
    /**
     * 上一次日志抓取量（判断服务是否挂掉）
     */
    public Integer lastLogNum = 0;

    @Override
    public void run() {
        String previousTime; // 上一条日志的打印时间
        String times;
        isRun = true;
        KafkaConsumer<String, String> consumer = null;
        List<JSONObject> list;
        MongoCollection<Document> collection = MongoTest.getCollection("api_log_" + time);
        try {
            consumer = KafkaCustomer.getConsumer();
            consumer.subscribe(Arrays.asList(topic));
            // 添加启动记录
            addLog(topic, "重启成功", null, null);
            while (isRun) {
                ConsumerRecords<String, String> records = consumer.poll(100); // 每100ms读取1次kafka
                for (ConsumerRecord<String, String> record : records) { // 遍历数据
                    String value = record.value();
                    int i = value.indexOf("{");
                    String timeStr = value.substring(0, 23);
                    times = value.substring(0, 10);
                    if (!DateUtil.isTime(times)) {
                        continue;
                    }
                    value = value.substring(i, value.length());
                    JSONObject obj = JSONObject.fromObject(value);
                    obj.put("partition", record.partition());
                    obj.put("offset", record.offset());
                    obj.put("time", timeStr);
                    obj.put("topic", topic);
                    logNum++;
                    previousTime = logMap.getTimeStr();
                    logMap.setLog(times, obj);
                    list = logMap.getList(times);
                    for (KafkaHandle handle : handles) {
                        handle.handle(logMap, topic, times);
                    }
//                    if (!times.equals(previousTime)) {
//                        if (logMap.getList(previousTime) != null) {
//                            add(previousTime, logMap.getList(previousTime), MongoTest.getCollection("api_log_" + previousTime));
//                        }
//                    }
                    if (list.size() == 50) { // 每50条执行一次插入
//                        add(times, list, collection);
                    }
                }
            }
        } catch (Exception e) {
            isRun = false;
            addLog(topic, "服务关闭", e.toString(), null);
        } finally {
            for (Map.Entry<String, List<JSONObject>> entry : logMap.getLogMap().entrySet()) {
                list = entry.getValue();
                if (null != list && list.size() > 0) {
                    add(entry.getKey(), list, MongoTest.test.getCollection("api_log_" + entry.getKey()));
                }
            }
            if (null != consumer) {
                consumer.close();
            }
            isRun = false;
            addLog(topic, "服务关闭 : " + logNum, null, null);
        }
    }

    public void stop() {
        logNum = 0;
        lastLogNum = 0;
        isRun = false;
    }

    public void add(String time, List<JSONObject> list, MongoCollection<Document> collection) {
        long start = System.currentTimeMillis();
        MongoTest.insertList(collection, list);
        long end = System.currentTimeMillis();
        LOG.info(topic + " : " + time + " ： 添加成功 ：" + logNum + " : " + (end - start));
    }

    /**
     * 添加重启日志
     * @param topic
     */
    public static void addLog(String topic, String msg, String error, Long costTime) {
        JSONObject obj = new JSONObject();
        obj.put("topic", topic);
        obj.put("time", DateUtil.dateToString(new Date(), "yyyy-MM-dd HH:mm:ss"));
        obj.put("info", msg);
        obj.put("error", error);
        obj.put("cost_time", costTime);
        MongoTest.insertOne(MongoTest.getCollection("kafka_start"), obj);
        LOG.info(topic + " : " + msg);
    }

    public static void main(String[] args) {
        List<JSONObject> list;
        LogMap logMap = new LogMap();
        Date time = DateUtil.calculateDate(new Date(), -1);
        String timeStr = DateUtil.dateToString(time, "yyyy-MM-dd");
        String times;
        for (int i = 0; i < 100; i++) {
//            if (i > 50) {
//                time = DateUtil.calculateDate(time, -1);
//                timeStr = DateUtil.dateToString(time, "yyyy-MM-dd");
//            }
            times = logMap.getTimeStr();
            logMap.setLog(timeStr, new JSONObject());
            list = logMap.getList(timeStr);
            if (!times.equals(timeStr)) {
                if (logMap.getList(times) != null) {
                    System.out.println(times + ":" + logMap.getList(times).size());
                }
            }
            if (list.size() >= 50) {
                System.out.println(timeStr + ":" + list.size());
                logMap.clean(timeStr);
            }
        }
        for (Map.Entry<String, List<JSONObject>> entry : logMap.getLogMap().entrySet()) {
            System.out.println(entry.getKey() + " : " + entry.getValue().size());
        }
    }

    public KafkaRunnable(String topic, List<KafkaHandle> handles) {
        this.topic = topic;
        this.handles = handles;
    }

    public String getTopic() {
        return topic;
    }

    public boolean isRun() {
        return isRun;
    }

    public String getTime() {
        return time;
    }
}

